package com.yfbao.poll.demo.service;

import cn.hutool.json.JSONUtil;
import com.yfbao.poll.demo.message.BasePollMessage;
import com.yfbao.poll.demo.model.AsyncLongPoll;
import com.yfbao.poll.demo.model.BasePoll;
import com.yfbao.poll.demo.model.SyncLongPoll;
import com.yfbao.poll.demo.vo.PollVo;
import com.yfbao.poll.demo.vo.UserMessageVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.LockSupport;


/**
 * 轮询业务逻辑处理类
 */
@Slf4j
@Component
public class PollingService {


    //保存轮询的请求（目前设计只支持同一用户单个客户端模式）
    private static final Map<String, List<BasePoll>> allWebMap = PollingCache.getAllWebMap();

    //长轮询开启的开关
    private static boolean longPollingCfg=  true;

    //默认的长轮询时长
    private static Integer defaultLongPollingTimeout = 30;

    //长轮询超时时间的偏移
    private static Integer delayOffset = 2;

    //预留：版本号
    private static String defTag = "1.2";

    /**
     * 处理及时消息
     * @param msg 即时接收到的用户消息
     * @return
     */
    public boolean messageHandlerTask(BasePollMessage msg){
        if(!longPollingCfg){
            //没有开启长轮询，走周期轮询的流程
            return false;
        }
        String loginName = msg.getLoginName();
        String msgType = msg.getMsgType();
        try {
            List<BasePoll> basePolls = allWebMap.get(loginName);
            if(basePolls ==null){
                return false;
            }
            Iterator<BasePoll> iterator = basePolls.iterator();
            BasePoll webPolling = null;
            while (iterator.hasNext()){
                webPolling = iterator.next();
                Map<String, Long> userMsgTypes = webPolling.getUserMsgTypes();
                if (userMsgTypes.containsKey(msgType)) {
                    Long lastMsgId = userMsgTypes.get(msgType);
                    if (lastMsgId !=null && lastMsgId>0 && (msg.getMsgId() < lastMsgId)) {
                        log.warn("webListen msgID:" + lastMsgId + " > currentMsgId:" + msg.getMsgId() + ",break this msg:" + loginName + "," + msgType);
                        return true;
                    }
                    break;
                }
            }
            if(webPolling ==null){
                return false;
            }
            boolean remove = basePolls.remove(webPolling);
            //如果用户监听请求中监听了对应事件类型，则执行下属逻辑
            //从轮询队列移除
            if(remove) {
                log.info("messageHandlerTask real-time processing:" + loginName + "," + msgType);
                Map<String, List<BasePollMessage>> msgMap = new HashMap<>();
                msgMap.put(msg.getMsgType(), Arrays.asList(msg));

                if (webPolling.isAsync()) {
                    //异步长轮询
                    AsyncLongPoll asyncLongPoll = (AsyncLongPoll) webPolling;
                    Map<String, Object> resMap = new HashMap<>();
                    resMap.put("longPollTimeout", asyncLongPoll.getTimeoutTime() + delayOffset);
                    resMap.put("message", msgMap);
                    resMap.put("messageLength",msgMap.size());
                    asyncLongPoll.sendResponse(resMap);
                } else {
                    //同步长轮询
                    SyncLongPoll syncLongPoll = (SyncLongPoll) webPolling;

                    Map<String, List<BasePollMessage>> userMessageMap = PollingCache.getUserMessageCacheMap();
//                    String cacheKey = loginName;
                    String cacheKey=loginName+"-"+msgType;
                    List<BasePollMessage> userMessageList = userMessageMap.get(cacheKey);
                    if (userMessageList == null) {
                        userMessageList =new CopyOnWriteArrayList<BasePollMessage>();
                        List<BasePollMessage> basePollMessages = userMessageMap.putIfAbsent(cacheKey, userMessageList);
                        if(basePollMessages !=null){
                            userMessageList = basePollMessages;
                        }
                    }
                    userMessageList.add(msg);

                    syncLongPoll.removeFutureTaskAndUnpark();
                }
                return true;
            }


        }catch (Throwable t){
            log.error("messageHandlerTask error: {}", t);
            return false;
        }
        return false;
    }

    /**
     * 核心轮询处理类
     * 如果长轮询不支持，则降为一般轮询
     * 业务；在长轮询等待前，先取判断一次是否已经有缓存的消息，有则立即返回；没有进入长轮询
     * @param request
     * @param response
     * @param reqVo
     */
    public void doPolling(HttpServletRequest request, HttpServletResponse response, String userName, PollVo reqVo){

        String longPollTimeout = reqVo.getLongPollTimeout();
        Long pollTimeOut = defaultLongPollingTimeout.longValue();
        if(longPollTimeout!=null){
            try {
                pollTimeOut = Long.valueOf(longPollTimeout);
            }catch (Throwable t){
                log.error("parse  pollTimeOut error",t);
            }
        }
        //设置长轮询的偏移
        pollTimeOut = pollTimeOut-delayOffset;

        List<UserMessageVo> valueList = reqVo.getPollMessage();
//        if(valueList ==null || valueList.isEmpty()){
//            log.error(currentUser.getLoginName()+",poll message params is null");
//            generate403Response(request,response,"poll message params is null");
//            return;
//        }

        //1、执行长轮询前先判断缓存是否已经有了相关消息，有则直接返回
        Map<String,Long> userMsgTypes = new HashMap<>();
        if(valueList!=null && !valueList.isEmpty()){
            for (UserMessageVo userMessageVo : valueList) {
                userMsgTypes.put(userMessageVo.getMsgType(),userMessageVo.getMsgId());
                if(userMessageVo.getMsgId() == null){
                    userMsgTypes.put(userMessageVo.getMsgType(),-1L);
                }
            }
        }

        Map<String, List<BasePollMessage>> cacheUserMsgs = getUserMessagesFromCache(userMsgTypes, userName);
        Map<String,Object> resMap =new HashMap<>();
        if(!cacheUserMsgs.isEmpty()){
            if(longPollingCfg && longPollTimeout!=null){
                //按照长轮询返回
                resMap.put("longPollTimeout",pollTimeOut+delayOffset);
            }
//            else{
//                log.info("finish short poll");
//            }
            resMap.put("message",cacheUserMsgs);
            resMap.put("messageLength",cacheUserMsgs.size());
            generateResponse(request,response,resMap);
            return;
        }
        //2。执行长轮询
        if (longPollingCfg && longPollTimeout!=null ) {
            //清除存在的历史轮询
            handlerHistoryRequest(userName,userMsgTypes,pollTimeOut);
            //用于启动异步工作线程,进入异步模式,调用业务处理线程进行业务处理
            boolean asyncSupported = request.isAsyncSupported();
            if(asyncSupported){
                request.startAsync(request, response);
                if ( request.isAsyncStarted()) {
                    doLongPolling(request,response, userMsgTypes,userName, pollTimeOut);
                    return;
                }
            }else {
                Thread thread = Thread.currentThread();
                SyncLongPoll syncLongPoll = new SyncLongPoll(thread, userMsgTypes, userName, pollTimeOut, defTag);
                syncLongPoll.addWebPollRequest();
                LockSupport.parkUntil(System.currentTimeMillis()+pollTimeOut*1000);
                cacheUserMsgs = getUserMessagesFromCache(userMsgTypes, userName);
                //执行同步唤起后的流程
                resMap.put("longPollTimeout",pollTimeOut+delayOffset);
                resMap.put("message",cacheUserMsgs);
                resMap.put("messageLength",cacheUserMsgs.size());
                generateResponse(request,response,resMap);
                return;
            }
        }
        //非长轮询时，按照 一般轮询返回
        resMap.put("message",cacheUserMsgs);
        resMap.put("messageLength",cacheUserMsgs.size());
        generateResponse(request,response,resMap);
        log.info("finish short poll");
    }


    /**
     * 如果有新的用户监听相同内容的长轮询请求进来，则立即处理掉该用户的历史长轮询请求
     * @param userName
     */
    private void handlerHistoryRequest(String userName, Map<String, Long> userMsgTypes, long pollTimeOut) {
        List<BasePoll> basePolls = allWebMap.get(userName);
        if (basePolls == null) {
            return;
        }
        Iterator<BasePoll> iterator = basePolls.iterator();
        while (iterator.hasNext()) {
            BasePoll basePoll = iterator.next();
            Map<String, Long> listenUserMsgTypes = basePoll.getUserMsgTypes();
            if (listenUserMsgTypes != null) {
                //如果新的请求监听包含老请求的监听内容，则立即反馈老的监听请求
                boolean isContinue = false;
                for (String s : listenUserMsgTypes.keySet()) {
                    if (!userMsgTypes.keySet().contains(s)) {
                        isContinue = true;
                        break;
                    }
                }
                if (isContinue) {
                    continue;
                }
            }
            try {
                boolean remove = basePolls.remove(basePoll);
                if (remove) {
                    removePoll(basePoll, userName, pollTimeOut);
                }
            } catch (Throwable t) {
                log.error("remove poll error", t);
            }
            log.info("remove old poll," + userName);
        }
    }

    private void removePoll(BasePoll poll, String userName, long pollTimeOut){
        Map<String, Long> userMsgTypes = poll.getUserMsgTypes();
        if(poll.isAsync()){
            //异步非阻塞长轮询
            AsyncLongPoll asyncLongPoll = (AsyncLongPoll)poll;

            Map<String, List<BasePollMessage>> cacheUserMsgs = getUserMessagesFromCache(userMsgTypes, userName);
            Map<String,Object> resMap =new HashMap<>();
            resMap.put("longPollTimeout",pollTimeOut+delayOffset);
            resMap.put("message",cacheUserMsgs);
            resMap.put("messageLength",cacheUserMsgs.size());
            asyncLongPoll.sendResponse(resMap);
        }else {
            //阻塞式长轮询
            SyncLongPoll syncLongPoll = (SyncLongPoll)poll;
            syncLongPoll.removeFutureTaskAndUnpark();
        }
    }


    /**
     * 目次此函数兼容设计，
     * 如果messageTypes为null时，表示全部消息都响应，否则只响应messageTypes中监听的消息
     * @param messageTypes
     * @param userName
     * @return
     */
    public Map<String, List<BasePollMessage>> getUserMessagesFromCache(Map<String, Long> messageTypes, String userName) {

        Map<String, List<BasePollMessage>> userMsgs = new HashMap<>();

        Set<String> keySet = messageTypes.keySet();
        for (String msgType : keySet) {
            String cacheKey = userName + "-" + msgType;
            List<BasePollMessage> basePollMessages = PollingCache.getUserMessageCacheMap().get(cacheKey);
            ArrayList<BasePollMessage> sendMessages = new ArrayList<>();
            if (basePollMessages == null || basePollMessages.isEmpty()) {
                continue;
            }
            Long aLong = messageTypes.get(msgType);
            Iterator<BasePollMessage> iterator = basePollMessages.iterator();
            while (iterator.hasNext()) {
                BasePollMessage next = iterator.next();
                if (aLong != null && next.getMsgId() > aLong) {
                    sendMessages.add(next);
                } else {
                    log.info("remove old message, currentMsgId:" + aLong + ",old:" + next.getMsgId());
                }
                basePollMessages.remove(next);
            }
            if(!sendMessages.isEmpty()){
                userMsgs.put(msgType, sendMessages);
            }
        }

        return userMsgs;
    }

    //执行异步长轮询
    public void doLongPolling(HttpServletRequest request, HttpServletResponse response,Map<String,Long> userMsgTypes,String userName,long timeoutTime){

        // 释放http连接，转为异步
        /**
         * 1 获取AsyncContext，对异步执行的上下文提供支持，可以透过AsyncContext的getRequest() 、 getResponse()方法取得Request、Response对象
         * 2  客户端的响应将暂缓至，调用AsyncContext的complete()方法或dispatch()为止，前者表示回应完成，后者表示将响应调派给指定的URL
         * 3 使用异步处理方式，web容器的请求处理线程释放了，可以服务其他的请求处理。但是该Request的处理并没有结束，
         *   在使用AsyncContext的complete或者dispatch完成后，这个request的处理才结束。
         */
        request.getServletContext();
        final AsyncContext context = request.getAsyncContext();
        context.setTimeout(0L);
        PollingExecutor.executeNormalTask(new AsyncLongPoll(context,userMsgTypes,userName,timeoutTime,defTag,this,delayOffset));
    }


    /**
     * 即时返回轮询请求
     * @param request
     * @param response
     * @param changedGroups
     */
    private void generateResponse(HttpServletRequest request, HttpServletResponse response, Map<String,Object> changedGroups) {
        if (null == changedGroups) {
            return;
        }
        try {
            //TODO：----
            response.setHeader("Pragma", "no-cache");
            response.setDateHeader("Expires", 0);
            response.setHeader("Cache-Control", "no-cache,no-store");
            response.setStatus(HttpServletResponse.SC_OK);
            if(!changedGroups.isEmpty()){
                String res = JSONUtil.toJsonStr(changedGroups);
                response.getWriter().println(res);
            }else {
                response.getWriter().println("no");
            }
            response.getWriter().flush();
        } catch (Exception ex) {
            log.error(ex.toString(), ex);
        }
    }

    private  void generate403Response(HttpServletRequest request, HttpServletResponse response, String message) {
        //todo:
        try {
            // Disable cache.
            response.setHeader("Pragma", "no-cache");
            response.setDateHeader("Expires", 0);
            response.setHeader("Cache-Control", "no-cache,no-store");
            response.setStatus(HttpServletResponse.SC_FORBIDDEN);
            response.getWriter().println(message);
            response.getWriter().flush();
        } catch (Exception ex) {
            log.error(ex.toString(), ex);
        }
    }

    private void generate503Response(HttpServletRequest request, HttpServletResponse response, String message) {
        //todo:
        try {
            // Disable cache.
            response.setHeader("Pragma", "no-cache");
            response.setDateHeader("Expires", 0);
            response.setHeader("Cache-Control", "no-cache,no-store");
            response.setStatus(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
            response.getWriter().println(message);
            response.getWriter().flush();
        } catch (Exception ex) {
            log.error(ex.toString(), ex);
        }
    }


//    public static void main(String[] args) {
//        CopyOnWriteArrayList<BasePoll> basePolls = new CopyOnWriteArrayList<>();
//        basePolls.add(new SyncLongPoll(null,null,null,0L,null));
//        Iterator<BasePoll> iterator = basePolls.iterator();
//        while (iterator.hasNext()){
//            BasePoll basePoll = iterator.next();
//            Map<String, Long> listenUserMsgTypes = basePoll.getUserMsgTypes();
//            boolean remove = basePolls.remove(basePoll);
////            if(remove){
////                System.out.println("remove poll error");
////            }
//            System.out.println("remove old poll,");
//        }
//    }
}
